package com.blue.rpc.core.registry;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.blue.rpc.client.RpcClient;
import com.blue.rpc.util.RpcConst;

/**
 * ZooKeeper 服务发现
 * 
 * @author zhengj
 * @since 1.0 2017年3月13日
 */
public class ZooKeeperDiscovery implements Discovery
{
	private static Logger logger = LoggerFactory.getLogger(ZooKeeperDiscovery.class);
	
	private String address;
	private RpcClient rpcClient;
	private ZooKeeper zooKeeper;
	private ServiceWatcher serviceWatcher;
	
	public ZooKeeperDiscovery(String address, RpcClient rpcClient)
	{
		this.address = address;
		this.serviceWatcher = new ServiceWatcher(this);
		this.rpcClient = rpcClient;
	}

	@Override
	public void afterPropertiesSet() throws Exception
	{
		this.connect();
		this.setRoot();
		this.updateService();
	}

	@Override
	public void destroy() throws Exception
	{
		if (zooKeeper != null)
		{
			zooKeeper.close();
		}
	}
	
	// 连接 ZooKeeper
	private void connect()
	{
		final CountDownLatch latch = new CountDownLatch(1);
		try
		{
			zooKeeper = new ZooKeeper(address, RpcConst.TIMEOUNT, (e) ->
			{
				// 连接成功
				if (e.getState() == Watcher.Event.KeeperState.SyncConnected)
				{
					logger.info("连接 ZooKeeper 成功：{}", address);
					latch.countDown();
				}
			});
			latch.await();
		}
		catch (IOException e)
		{
			logger.error("连接 ZooKeeper 失败：" + address, e);
		}
		catch (InterruptedException e)
		{
			e.printStackTrace();
		}
	}
	// 设置 ZooKeeper 根路径
	private void setRoot()
	{
		try
		{
			Stat stat = zooKeeper.exists(RpcConst.RPC_ROOT, false);
			if (stat == null)
			{
				String path = zooKeeper.create(RpcConst.RPC_ROOT, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				logger.info("创建 ZooKeeper 根路径：{}", path);
			}
		}
		catch (KeeperException | InterruptedException e)
		{
			logger.error("创建 ZooKeeper 根路径有错误：", e);
		}
	}
	
	@Override
	public void updateService()
	{
		Map<String, List<String>> serviceMap = new HashMap<>();
		try
		{
			List<String> serviceList = zooKeeper.getChildren(RpcConst.RPC_ROOT, serviceWatcher);
			for (String service : serviceList)
			{
				List<String> hostList = zooKeeper.getChildren(RpcConst.RPC_ROOT + "/" + service, serviceWatcher);
				logger.info("发现服务：{}－{}", service, hostList);
				serviceMap.put(service, hostList);
			}
		}
		catch (KeeperException | InterruptedException e)
		{
			logger.error("获取 ZooKeeper 服务列表有错误：", e);
		}
		
		rpcClient.updateConnect(serviceMap);
	}
	
}
